#!/usr/bin/env python
# -*- encoding: utf-8 -*-
'''
@File    :   sensor_test.py
@Time    :   2024/05/04 09:54:12
@Author  :   Jim 
@Version :   1.0.1
@Contact :   jim@yi-win.com
@License :   (C)Copyright 2021-2025, 奕枫仪器
@Desc    :   测试  结合单元测试 unittest
'''

from tkinter import *
from tkinter import ttk
import tkinter.messagebox
import threading
import serial
import serial.tools.list_ports
import  sys
import time
import logging
from isensor import CMD_REPLY, isensor ,sensor_hex ,sensor_ascii, sensor_enviroFlu
from HTMLTestRunner_PY3 import HTMLTestRunner
from pathlib import Path,PurePath
import unittest

logger = logging.getLogger(__name__)
logger.setLevel(level=logging.INFO)

formatter = logging.Formatter('%(asctime)s - %(name)s [%(lineno)s] - %(levelname)s - %(message)s')

console_handler = logging.StreamHandler()  # 流句柄
console_handler.setLevel(logging.INFO)
console_handler.setFormatter(formatter)
logger.addHandler(console_handler)

REPORT_FNAME = "report.html"
TEST_PORT = "COM1"
TEST_BAUD = "9600"


class Uart(object):
    def __init__(self,  ):
        self.ser  = None  
        self.port = None      
        self.baudrate = None      
        self.databit = None      
        self.stopbit = None      
        self.parity = None  
        
        self.rcv_timeout = None  
        self.rcv_ok = False  

        self.sensor:isensor         = None
        self.bufmachine:buf_machine =buf_machine()
        self.thread:threading.Thread = None

    def uart_set_buf_machine(self,machine ):
        self.bufmachine =  machine
        self.bufmachine.clean()

    def uart_set_sensor(self,sensor:isensor):
        self.sensor =  sensor

    def uart_thread_recv(self, func):
        """
        串口数据接受线程
        """
        readout = b''
        if self.ser.is_open is False:
            self.ser.open()     
            
        while(True):
            # logger.info("threading.....3")
            cnt = 0
            try:
                cnt =self.ser.in_waiting
            except Exception:
                pass
            if cnt :
                logger.info(f"threading.....{cnt}")
                readout =  self.ser.read(cnt) 
                func(readout)
            time.sleep(0.02)    

    def  uart_open(self, port_, baudrate_):
        if self.ser is not None:
            self.ser.close()
            logger.info(f"uart open fail")
        self.ser = serial.Serial(
                                port = port_,
                                baudrate=int(baudrate_),
                                parity=serial.PARITY_NONE,
                                stopbits=serial.STOPBITS_ONE,
                                bytesize=serial.EIGHTBITS )            

        if self.ser.isOpen():
            self.thread = threading.Thread( target=self.uart_thread_recv, args=(self.bufmachine.process,) )
            self.thread.setDaemon(True)
            self.thread.start()
            return True
        return False 

    def uart_sent(self, cmd_seq):
        self.rcv_ok = False
        self.bufmachine.clean()
        snd_ = b''
        if self.sensor.cmd_reply.hex_ascii == 1:
            snd_ =  bytes.fromhex(self.sensor.cmd_reply.cmd[cmd_seq])
        if self.sensor.cmd_reply.hex_ascii == 2:
            snd_  = bytes(self.sensor.cmd_reply.cmd[cmd_seq], encoding='utf8')
        if self.ser.isOpen():
            self.ser.write(snd_)

    def uart_close(self, ):
        self.ser.close()

class buf_machine(object):
    # 构造方法
    def __init__(self, ):
        self.hex_ascii = 1
        self.buf = b''
        self.res = ""
        self.rcv_ok = False
        self.validate = None
        pass

    def set_hex_ascii (self,hex_ascii):
        self.hex_ascii = hex_ascii
        pass

    def set_validate (self,validate):
        self.validate = validate
        pass

    def clean (self, ):
        self.buf = b''
        self.res = ""
        self.rcv_ok = False
        pass

    def get_res (self, ):
        return self.res

    def get_buf (self, ):
        return self.buf

        
    def process(self,_byte:bytes):
        logger.info(f'process......... {_byte}')
        self.buf += _byte
        tmp= ""
        if self.hex_ascii == 1:
            tmp = _byte.hex()
        if self.hex_ascii == 2 and  _byte != b'\r'  and _byte != b'\n' :
            tmp = _byte.decode()
        self.res += tmp
        return self.res
        pass

class TestTest(unittest.TestCase):
    """ 测试HTMLTestRunner """
    def setUp(self):
        self.uart =  Uart() 
        
        self.bufmachine:buf_machine = buf_machine()
        self.uart.uart_set_buf_machine(self.bufmachine)
        self.bufmachine.clean()    
        
        self.sensor = sensor_hex("sensorHex")
        # self.uart.bufmachine.set_validate(self.uart.sensor.validate)
        self.cmd_seq = 0
        # self.uart.uart_open(TEST_PORT,TEST_BAUD)
        
    def get_cmd_seq(self):
        lens = len(self.sensor.cmd_reply.cmd)
        if lens == 0:
            return
        import random
        self.cmd_seq = random.randint(0, lens-1)
        return self.cmd_seq

    # def test_sensor_hex(self):
    #     """ sensor_hex 测试 """
    #     logger.info(f"test_sensor_hex .... ")
    #     self.sensor = sensor_hex("sensorHex")
    #     self.uart.uart_set_sensor(self.sensor)
    #     self.uart.bufmachine.set_validate(self.uart.sensor.validate)
    #     self.uart.sensor.handle()
    #     self.uart.uart_open(TEST_PORT,TEST_BAUD)
    #     self.get_cmd_seq( )
    #     self.uart.sensor.set_seq(self.cmd_seq)
 
    #     self.uart.uart_sent( self.cmd_seq )
    #     time.sleep( 2*self.sensor.timeout/1000 )
    #     res = self.uart.sensor.validate( self.uart.bufmachine.res, self.cmd_seq ) 
    #     rcv = self.uart.bufmachine.res
    #     PASSED = "PASS" if res == True else "FAIL"
    #     logger.info( f"test_sensor_hex ..[seq { self.cmd_seq}]  [cmd: {self.uart.sensor.cmd_reply.cmd[self.cmd_seq]} ]  [rcv:{rcv}]  -> [{PASSED}]" )
    #     print( f"test_sensor_hex ..[seq { self.cmd_seq}]  [cmd: {self.uart.sensor.cmd_reply.cmd[self.cmd_seq]} ]  [res  {res}]" )
    #     # self.uart.tread.stop()
    #     self.uart.uart_close()
    #     self.assertEqual(res, True ) 


    @unittest.skip
    def test_sensor_hex_sub(self):
        """ sensor_hex 测试 """
        self.sensor = sensor_hex("sensorHex" )
        self.uart.uart_set_sensor(self.sensor)
        self.uart.bufmachine.set_validate(self.uart.sensor.validate)
        self.uart.sensor.handle()
        self.uart.uart_open(TEST_PORT,TEST_BAUD)
        times = 5
        for i in range(times):
            self.get_cmd_seq( )
            self.uart.sensor.set_seq(self.cmd_seq)
            with self.subTest(msg="No. "+str(i),  seq =self.cmd_seq, cmd = self.uart.sensor.cmd_reply.cmd[self.cmd_seq], rcv=self.bufmachine.res):
                self.uart.uart_sent( self.cmd_seq )
                time.sleep( 2*self.sensor.timeout/1000 )
                res = self.uart.sensor.validate( self.uart.bufmachine.res, self.cmd_seq ) 
                logger.info( f"test_sensor_hex [{i}]   [res:{res}]   " )

                rcv = self.uart.bufmachine.res
                PASSED = "PASS" if res == True else "FAIL"
                logger.info( f"test_sensor_hex [{i}]  [seq:{ self.cmd_seq}]  [cmd: {self.uart.sensor.cmd_reply.cmd[self.cmd_seq]}] [rcv:{rcv}]  -> [{PASSED}]" )
                # print( f"test_sensor_hex [{i}]  [seq { self.cmd_seq}]  [cmd: {self.uart.sensor.cmd_reply.cmd[self.cmd_seq]} ]  [validate result : {res}]" )
                # self.uart.uart_close()
                self.assertEqual(res, True )

    def test_sensor_enviroflu_sub(self):
        """ sensor_enviroFlu 测试 """
        self.sensor = sensor_enviroFlu("enviroFlu" )
        self.uart.uart_set_sensor(self.sensor)
        self.uart.bufmachine.set_validate(self.uart.sensor.validate)
        self.uart.sensor.handle()
        self.uart.uart_open(TEST_PORT,TEST_BAUD)
        times = 5
        for i in range(times):
            self.get_cmd_seq( )
            self.uart.sensor.set_seq(self.cmd_seq)
            with self.subTest(msg="No. "+str(i), seq =self.cmd_seq, cmd = self.uart.sensor.cmd_reply.cmd[self.cmd_seq], rcv=self.bufmachine.res):
                self.uart.uart_sent( self.cmd_seq )
                time.sleep( 2*self.sensor.timeout/1000 )
                res = self.uart.sensor.validate( self.uart.bufmachine.res, self.cmd_seq ) 
                logger.info( f"test_sensor_hex [{i}]   [res:{res}]   " )

                rcv = self.uart.bufmachine.res
                PASSED = "PASS" if res == True else "FAIL"
                logger.info( f"test_sensor_hex [{i}]  [seq:{ self.cmd_seq}]  [cmd: {self.uart.sensor.cmd_reply.cmd[self.cmd_seq]}] [rcv:{rcv}]  -> [{PASSED}]" )
                # print( f"test_sensor_hex [{i}]  [seq { self.cmd_seq}]  [cmd: {self.uart.sensor.cmd_reply.cmd[self.cmd_seq]} ]  [validate result : {res}]" )
                # self.uart.uart_close()
                self.assertEqual(res, True )

    # @unittest.skip
    def test_a_plus_b(self):
        """ a + b = 3 这个用例应该通过"""
        print('1 + 2 = 3 Success')
        self.assertEqual(1+2, 3)

    def test_a_minus_b(self):
        """ a - b = 3 这个用例应该失败 """
        print('3 - 1 = 3 Fault')
        self.assertEqual(3-1, 3)


if __name__ == "__main__":
    print(" ********* Begin to  sensor test ********* ")
    print(" *** 请修改串口设置 TEST_PORT TEST_BAUD *** ")
    print(" *** isensor.py 添加传感器类用于测试模拟 *** ")
    print(" *** isensor.py 添加  cmd   reply      *** ")
    print(" *** TestTest 添加函数  作为测试用例     *** ")
    print(" *** @unittest.skip 跳过暂且不用的case  *** ")
    print(" ***************************************** ")
    
    curr_path       = Path()
    report_name     = REPORT_FNAME
    report_fpath    = curr_path.joinpath( report_name)
    
    report_sensor_name  = "HEX"
    report_sensor_sn    = "snsnsn"
    report_man          = "奕枫"
    report_title    = "奕枫仪器 "+report_sensor_name+ "传感器测试"
    report_desc     = "序列号：" +report_sensor_sn + "                            报告人：" + report_man

    
    # testsuite       = unittest.TestSuite()
    # testsuite.addTest(unittest.TestLoader().loadTestsFromTestCase(TestTest))

    # # testsuite.addTests([TestTest('test_a_plus_b'), TestTest('test_a_divide_c')])

    # with open( report_fpath, 'wb') as report:
    #     runner = HTMLTestRunner( stream=report, title=report_title, description=report_desc )
    #     runner.run( testsuite )
    
    count = 1    
    for i in range(count):
        report_name     = "report_"+str(i)+".html"
        if i == 0:
            report_name     = "report.html"
        report_fpath    = curr_path.joinpath( report_name)
        testsuite       = unittest.TestSuite()
        testsuite.addTest(unittest.TestLoader().loadTestsFromTestCase(TestTest))

        # testsuite.addTests([TestTest('test_a_plus_b'), TestTest('test_a_divide_c')])

        with open( report_fpath, 'wb') as report:
            runner = HTMLTestRunner( stream=report, title=report_title, description=report_desc )
            runner.run( testsuite )
    